package jmind.core.mq.kafka;

import java.util.Properties;

import jmind.base.lang.SourceProperties;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * http://www.cnblogs.com/likehua/p/3999538.html
 * http://www.aboutyun.com/thread-12882-1-1.html
 */
public class KafkaProducerUtill {

    private static final Logger logger = LoggerFactory.getLogger(KafkaProducerUtill.class);

    private KafkaProducerUtill() {
    };

    private static class KafkaServerUtilHolder {
        private static KafkaProducer<String, String> producer;

        static {
            try {
                /**
                 * bootstrap.servers=10.135.29.25:9093,10.135.29.31:9094
                   client.id=kafka-test
                 */
                Properties props = new Properties();
                props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
                        SourceProperties.getDataSource().getProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG));
                //  leader 成功 则成功
                props.put(ProducerConfig.ACKS_CONFIG, "1");
                props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
                        "org.apache.kafka.common.serialization.StringSerializer");
                props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
                        "org.apache.kafka.common.serialization.StringSerializer");
                props.put(ProducerConfig.CLIENT_ID_CONFIG,
                        SourceProperties.getDataSource().getProperty(ProducerConfig.CLIENT_ID_CONFIG));
                producer = new KafkaProducer<String, String>(props);
            } catch (Exception e) {
                logger.error("kafka config error!", e);
            }
        }
    }

    public static void sendMessage(final String topic, final String key, final String value) {
        try {
            KafkaServerUtilHolder.producer.send(new ProducerRecord<String, String>(topic, key, value));
            logger.info("kafkaProducerStatistic-topic:[{}],key:[{}]", topic, key);//观查一段时间以后会去掉。
        } catch (Exception e) {
            logger.error("send kafka message error!", e);

        }
    }
}